#include "rm_scan.h"

#include "rm_file_handle.h"
#include "bitmap.h" // For Bitmap operations
#include "errors.h" // For error handling

/**
 * @brief ³õÊ¼»¯file_handleºÍrid
 *
 * @param file_handle
 */
RmScan::RmScan(const RmFileHandle *file_handle) : file_handle_(file_handle) {
    // Todo:
    // ³õÊ¼»¯file_handleºÍrid£¨Ö¸ÏòµÚÒ»¸ö´æ·ÅÁË¼ÇÂ¼µÄÎ»ÖÃ£©

    // ³õÊ¼»¯ rid_ ÎªÎÞÐ§Öµ
    rid_ = {RM_NO_PAGE, -1};

    // ÕÒµ½µÚÒ»¸ö´æ·ÅÁË¼ÇÂ¼µÄÎ»ÖÃ
    // ´ÓÎÄ¼þµÄµÚÒ»¸ö¼ÇÂ¼Ò³Ãæ£¨Í¨³£ÊÇ page_no = 1£©¿ªÊ¼²éÕÒ
    // RM_FIRST_RECORD_PAGE ¶¨ÒåÔÚ rm_defs.h ÖÐ
    int current_page_no = RM_FIRST_RECORD_PAGE;
    while (current_page_no < file_handle_->file_hdr_.num_pages) {
        // »ñÈ¡µ±Ç°Ò³ÃæµÄ page handle
        // ×¢Òâ£ºfetch_page_handle »á pin Ò³Ãæ£¬ÐèÒªÔÚ next() »ò is_end() ÖÐ unpin
        RmPageHandle page_handle = file_handle_->fetch_page_handle(current_page_no);

        // ÔÚµ±Ç°Ò³ÃæÖÐ²éÕÒµÚÒ»¸öÒÑÕ¼ÓÃµÄ slot
        int slot_no = Bitmap::first_bit(true, page_handle.bitmap, file_handle_->file_hdr_.num_records_per_page);

        // ÊÍ·ÅÒ³Ãæ
        // ÕâÀïÖ»ÊÇ³õÊ¼»¯£¬²»Ó¦¸Ã±ê¼ÇÎªÔà£¬ÒòÎªÃ»ÓÐÐÞ¸ÄÒ³ÃæÄÚÈÝ
        // ¸¨Öúº¯Êý£º½«Ò³Ãæ±ê¼ÇÎªÔà²¢È¡Ïû¹Ì¶¨ (ÕâÀï²»±ê¼ÇÎªÔà)
        if (!file_handle_->buffer_pool_manager_->UnpinPage(page_handle.page->GetPageId(), false)) {
             // ¼ÇÂ¼¾¯¸æ»òÅ×³ö´íÎó£¬Èç¹û unpin Ê§°Ü
             LOG_WARN("Failed to unpin page during RmScan initialization.");
        }


        if (slot_no != file_handle_->file_hdr_.num_records_per_page) {
            // ÕÒµ½ÁËµÚÒ»¸ö¼ÇÂ¼
            rid_ = {current_page_no, slot_no};
            return;
        }
        // µ±Ç°Ò³ÃæÃ»ÓÐ¼ÇÂ¼£¬¼ÌÐø²éÕÒÏÂÒ»¸öÒ³Ãæ
        current_page_no++;
    }

    // Èç¹û±éÀúÍêËùÓÐÒ³Ãæ¶¼Ã»ÓÐÕÒµ½¼ÇÂ¼£¬Ôò½« rid_ ÉèÎªÎÄ¼þÄ©Î²±êÊ¶
    rid_ = {RM_NO_PAGE, -1};
}

/**
 * @brief ÕÒµ½ÎÄ¼þÖÐÏÂÒ»¸ö´æ·ÅÁË¼ÇÂ¼µÄÎ»ÖÃ
 */
void RmScan::next() {
    // Todo:
    // ÕÒµ½ÎÄ¼þÖÐÏÂÒ»¸ö´æ·ÅÁË¼ÇÂ¼µÄ·Ç¿ÕÏÐÎ»ÖÃ£¬ÓÃrid_À´Ö¸ÏòÕâ¸öÎ»ÖÃ

    // Èç¹ûÒÑ¾­µ½´ïÎÄ¼þÄ©Î²£¬ÔòÎÞÐè¼ÌÐø
    if (is_end()) {
        return;
    }

    int current_page_no = rid_.page_no;
    int current_slot_no = rid_.slot_no;

    // »ñÈ¡µ±Ç°Ò³ÃæµÄ page handle
    // ×¢Òâ£ºFetchPage »á pin Ò³Ãæ£¬ÐèÒªÔÚº¯Êý½áÊøÊ± unpin
    RmPageHandle page_handle = file_handle_->fetch_page_handle(current_page_no);

    // ÔÚµ±Ç°Ò³ÃæÖÐ²éÕÒÏÂÒ»¸öÒÑÕ¼ÓÃµÄ slot
    int next_slot_no = Bitmap::next_bit(true, page_handle.bitmap, file_handle_->file_hdr_.num_records_per_page, current_slot_no);

    // ÊÍ·Åµ±Ç°Ò³Ãæ
    // ÕâÀï²»±ê¼ÇÎªÔà£¬ÒòÎªÃ»ÓÐÐÞ¸ÄÒ³ÃæÄÚÈÝ
    if (!file_handle_->buffer_pool_manager_->UnpinPage(page_handle.page->GetPageId(), false)) {
        LOG_WARN("Failed to unpin page during RmScan::next().");
    }

    if (next_slot_no != file_handle_->file_hdr_.num_records_per_page) {
        // ÔÚµ±Ç°Ò³ÃæÕÒµ½ÁËÏÂÒ»¸ö¼ÇÂ¼
        rid_ = {current_page_no, next_slot_no};
    } else {
        // µ±Ç°Ò³ÃæËùÓÐ slot ¶¼ÒÑ±éÀú£¬²éÕÒÏÂÒ»¸öÒ³Ãæ
        current_page_no++;
        while (current_page_no < file_handle_->file_hdr_.num_pages) {
            RmPageHandle next_page_handle = file_handle_->fetch_page_handle(current_page_no);
            int first_slot_in_next_page = Bitmap::first_bit(true, next_page_handle.bitmap, file_handle_->file_hdr_.num_records_per_page);

            // ÊÍ·ÅÒ³Ãæ
            if (!file_handle_->buffer_pool_manager_->UnpinPage(next_page_handle.page->GetPageId(), false)) {
                LOG_WARN("Failed to unpin page during RmScan::next() (next page search).");
            }

            if (first_slot_in_next_page != file_handle_->file_hdr_.num_records_per_page) {
                // ÔÚÏÂÒ»¸öÒ³ÃæÕÒµ½ÁË¼ÇÂ¼
                rid_ = {current_page_no, first_slot_in_next_page};
                return;
            }
            current_page_no++;
        }
        // ±éÀúÍêËùÓÐÒ³Ãæ¶¼Ã»ÓÐÕÒµ½¼ÇÂ¼£¬µ½´ïÎÄ¼þÄ©Î²
        rid_ = {RM_NO_PAGE, -1};
    }
}

/**
 * @brief ? ÅÐ¶ÏÊÇ·ñµ½´ïÎÄ¼þÄ©Î²
 */
bool RmScan::is_end() const {
    // Todo: ÐÞ¸Ä·µ»ØÖµ
    // µ± rid_ µÄ page_no Îª RM_NO_PAGE Ê±£¬±íÊ¾µ½´ïÎÄ¼þÄ©Î²
    return rid_.page_no == RM_NO_PAGE;
}

/**
 * @brief RmScanÄÚ²¿´æ·ÅµÄrid
 */
Rid RmScan::rid() const {
    // Todo: ÐÞ¸Ä·µ»ØÖµ
    return rid_;
}

